Node.js Streams: Transform, Duplex e Backpressure

Node.js Streams são um dos recursos mais importantes da plataforma para lidar com dados de forma eficiente, especialmente quando o volume é grande ou quando os dados chegam aos poucos (I/O). Em vez de carregar tudo na memória e só então processar, Streams permitem ler, transformar e escrever em partes, mantendo a aplicação responsiva e com consumo de recursos previsível.

Este artigo foca em três pontos que costumam gerar dúvidas mesmo em projetos maduros:

Duplex streams: quando o mesmo objeto pode ler e escrever.
Transform streams: um tipo especial de Duplex que “transforma” o dado no meio do caminho.
Backpressure: o mecanismo que evita que o produtor de dados “atropеле” o consumidor, prevenindo filas gigantes e uso excessivo de memória.

A meta aqui é entender o funcionamento e aplicar padrões seguros e performáticos no dia a dia.

O que são Streams no Node.js (contexto rápido)

No Node.js, um stream é uma abstração para lidar com dados em fluxo. Existem quatro categorias principais:

Readable: emite dados (ex.: fs.createReadStream).
Writable: recebe dados (ex.: fs.createWriteStream).
Duplex: é Readable e Writable ao mesmo tempo (ex.: net.Socket).
Transform: é Duplex, mas com transformação entre entrada e saída (ex.: zlib.createGzip()).

Na prática, Streams são fundamentais em:

processamento de logs e arquivos grandes,
pipelines de compressão/criptografia,
uploads e downloads,
proxies, gateways e integrações com APIs,
parsing incremental (CSV/JSONL), ETL e processamento de eventos.

Duplex Stream: leitura e escrita no mesmo objeto

Um Duplex stream implementa duas interfaces: Writable (entrada) e Readable (saída). Um exemplo comum é um socket TCP: você escreve bytes para enviar e lê bytes para receber.

Quando usar Duplex

Use Duplex quando:

a entidade tem canal de entrada e saída simultâneos;
a leitura e a escrita têm lógicas separadas (não necessariamente uma transforma a outra);
você está criando uma abstração de transporte (ex.: túnel, proxy, multiplexador).

Exemplo didático de Duplex customizado

A seguir, um Duplex que recebe dados e, de forma independente, emite “ticks” periódicos. Perceba que a leitura (_read) e a escrita (_write) não são a mesma coisa.

import { Duplex } from "node:stream";

class TickDuplex extends Duplex {
  constructor(options = {}) {
    super({ ...options });
    this._counter = 0;
    this._interval = setInterval(() => {
      const chunk = `tick:${++this._counter}\n`;
      const canPush = this.push(chunk);
      // Se push retornar false, o consumidor está lento.
      // Aqui poderíamos pausar o intervalo até o 'drain' do lado readable,
      // mas em Duplex isso exige controle adicional (ver seção de backpressure).
      if (!canPush) {
        // Simplesmente não fazemos nada aqui para manter o exemplo curto.
      }
    }, 1000);
  }

  _read() {
    // Chamado quando o consumidor quer mais dados.
    // Neste exemplo, o push é dirigido pelo setInterval.
  }

  _write(chunk, encoding, callback) {
    // Recebe dados do lado writable
    const input = chunk.toString();
    // Poderíamos processar/rotear. Aqui só registramos.
    process.stderr.write(`recebido: ${input}`);
    callback();
  }

  _final(callback) {
    clearInterval(this._interval);
    callback();
  }
}

// Uso
const d = new TickDuplex();
d.on("data", (c) => process.stdout.write(c));
d.write("hello\n");

Ponto-chave: Duplex não implica transformação automática. Ele apenas permite leitura e escrita coexistirem.

Transform Stream: Duplex com transformação

Um Transform stream é um Duplex em que o que entra (Writable) é transformado e então sai (Readable). Esse tipo é ideal para pipelines:

normalização de texto,
compressão,
criptografia,
parsing e serialização incremental,
filtragem e enriquecimento de dados.

Como o Transform funciona

Você implementa principalmente:

_transform(chunk, encoding, callback): recebe um pedaço e gera saída via this.push(...).
opcionalmente _flush(callback): para emitir dados finais quando a entrada termina.

Exemplo: transformar linhas (uppercase) com controle por chunk

Exemplo Didático De Duplex Customizado Node.Js

import { Transform } from "node:stream";

class UppercaseTransform extends Transform {
  constructor(options = {}) {
    super({ ...options });
    this._buffer = "";
  }

  _transform(chunk, encoding, callback) {
    // Streams podem quebrar texto no meio; por isso, juntamos em buffer.
    this._buffer += chunk.toString("utf8");

    const lines = this._buffer.split("\n");
    this._buffer = lines.pop(); // última pode estar incompleta

    for (const line of lines) {
      this.push(line.toUpperCase() + "\n");
    }

    callback();
  }

  _flush(callback) {
    if (this._buffer) {
      this.push(this._buffer.toUpperCase());
    }
    callback();
  }
}

Esse padrão (buffer + split) é comum em processamento de logs e arquivos texto, e evita um erro típico: assumir que cada chunk corresponde a uma linha completa.

Backpressure: o mecanismo que evita estouro de memória

Backpressure é o controle de fluxo entre quem produz dados e quem consome. Em termos simples:

se o destino (Writable) está lento,
o Node.js sinaliza isso,
e a origem (Readable/Transform) deve desacelerar.

Sem backpressure bem respeitado, um pipeline pode:

acumular buffers gigantes em RAM,
aumentar latência,
elevar custo operacional,
e em cenários extremos derrubar o processo (OOM).

Como o Node.js sinaliza backpressure

Do lado Writable:

writable.write(chunk) retorna true se o buffer interno ainda comporta mais dados.
retorna false quando o buffer atingiu o limite (highWaterMark).
quando o buffer esvazia o suficiente, o stream emite o evento drain.

Do lado de pipelines, usar pipe()/pipeline() já aplica backpressure automaticamente na maioria dos casos.

Padrão seguro: prefira `pipeline()` para compor Streams

Para conectar streams com tratamento robusto de erros e fechamento correto, o recomendado é pipeline().

Exemplo: ler arquivo grande, transformar e gravar em outro:

import { createReadStream, createWriteStream } from "node:fs";
import { pipeline } from "node:stream/promises";

import { Transform } from "node:stream";

class UppercaseTransform extends Transform {
  _transform(chunk, enc, cb) {
    this.push(chunk.toString("utf8").toUpperCase());
    cb();
  }
}

await pipeline(
  createReadStream("entrada.txt"),
  new UppercaseTransform(),
  createWriteStream("saida.txt")
);

O pipeline():

conecta os streams com backpressure,
propaga erros corretamente,
encerra recursos ao final (ou em falha),
reduz a chance de vazamentos de file descriptor e listeners.

Quando backpressure falha: o anti-padrão do `data` + `write` sem controle

Um erro comum é fazer:

readable.on("data", (chunk) => {
  writable.write(chunk);
});

Isso pode ignorar backpressure dependendo de como você trata o retorno do write().

O padrão correto (manual) exige:

verificar o retorno do write();
se retornar false, pausar o readable;
retomar no drain.

Exemplo correto:

readable.on("data", (chunk) => {
  const ok = writable.write(chunk);
  if (!ok) readable.pause();
});

writable.on("drain", () => {
  readable.resume();
});

readable.on("end", () => writable.end());

Na prática, isso é exatamente o tipo de detalhe que pipe() e pipeline() já resolvem, por isso são preferíveis.

`highWaterMark`: ajuste fino de memória e throughput

O highWaterMark define o tamanho do buffer interno que um stream tolera antes de começar a sinalizar backpressure.

Em Readable, é o quanto ele tenta manter “pré-lido” em memória.
Em Writable, é o quanto aceita enfileirar antes de write() retornar false.

Exemplo de ajuste (leitura e escrita):

import { createReadStream, createWriteStream } from "node:fs";

const readable = createReadStream("grande.bin", { highWaterMark: 1024 * 1024 }); // 1 MB
const writable = createWriteStream("copia.bin", { highWaterMark: 1024 * 256 }); // 256 KB

readable.pipe(writable);

Como escolher?

highWaterMark maior tende a aumentar throughput em I/O rápido, mas cresce uso de RAM.
highWaterMark menor reduz picos de memória, mas pode aumentar overhead e chamadas de sistema.

Em sistemas com múltiplos streams concorrentes (ex.: processando centenas de arquivos), essa configuração vira um ponto real de capacidade.

Implicações de Cyber Segurança: streams também podem virar vetor de risco

Mesmo sendo uma ferramenta de performance, Node.js Streams têm impacto direto em segurança e confiabilidade:

Negação de serviço (DoS) por memória: ignorar backpressure ao processar uploads pode permitir que um cliente force buffers a crescerem.
Zip bombs e decompress bombs: pipelines com zlib podem expandir dados muito além do esperado. Mitigue com limites (tamanho total, tempo, contagem de bytes).
Parsing incremental inseguro: um Transform que acumula buffer sem limites (ex.: esperando um delimitador que nunca chega) pode crescer indefinidamente. Estratégia: impor limite máximo de buffer e abortar quando exceder.
Tratamento de erro incompleto: não capturar erros em cada stream pode deixar conexões abertas e recursos pendurados. pipeline() reduz esse risco.

Para processamento de dados de origem externa (uploads, integrações, mensagens), trate streams como fronteira de segurança: limite, valide, registre e aplique timeouts.

Resumo prático

Duplex: lê e escreve; útil quando entrada e saída coexistem, mas não necessariamente se transformam.
Transform: um Duplex que converte dados; ideal para pipelines (normalizar, filtrar, comprimir, criptografar).
Backpressure: mecanismo central para performance e estabilidade; evita bufferização infinita.
Use pipeline() como padrão para compor streams com backpressure e erros bem tratados.
Evite o anti-padrão on('data') + write() sem pause()/resume() baseado em drain.
Considere limites e validações: Streams mal controlados podem amplificar riscos de DoS e consumo de memória.

Se você domina esses três pilares (Duplex, Transform e backpressure), passa a escrever pipelines mais rápidos, econômicos e previsíveis — e isso é exatamente o que normalmente separa um serviço Node.js estável de um que “engasga” em produção.