Web Streams API no Node.js

A Web Streams API no Node.js oferece interfaces padronizadas para produzir, transformar e consumir dados em fluxo. Ela usa classes como ReadableStream, WritableStream e TransformStream, também disponíveis em navegadores e runtimes de edge.

O Node.js já possui streams tradicionais há muitos anos. As Web Streams não substituem automaticamente essas APIs, mas facilitam interoperabilidade com fetch(), Response, Request e bibliotecas que seguem padrões da Web. Com adaptadores, é possível conectar os dois ecossistemas.

Neste guia, você aprenderá a criar streams legíveis e graváveis, controlar backpressure, usar readers, writers, piping, transformações, cancelamento, byte streams, conversão para streams do Node.js e práticas para evitar vazamentos ou consumo excessivo de memória.

O que são Web Streams?

Web Streams são APIs definidas em padrões da plataforma Web. A documentação oficial de Web Streams no Node.js mostra classes e adaptadores disponíveis. A documentação da Streams API na MDN explica o modelo compartilhado com navegadores.

Para o modelo tradicional do runtime, consulte Streams no Node.js. Para dados binários, veja Buffer no Node.js.

ReadableStream básico

const stream = new ReadableStream({
  start(controller) {
    controller.enqueue('primeiro');
    controller.enqueue('segundo');
    controller.close();
  }
});

O controller adiciona chunks e encerra o fluxo. Os chunks podem ser strings, objetos ou bytes, dependendo do contrato.

Consumindo com reader

const reader = stream.getReader();

try {
  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    console.log(value);
  }
} finally {
  reader.releaseLock();
}

Um stream bloqueado por um reader não pode ser consumido simultaneamente por outro reader comum.

Iteração assíncrona

for await (const chunk of stream) {
  console.log(chunk);
}

O suporte depende da versão. Iteração assíncrona simplifica o consumo e mantém processamento sequencial.

Produção sob demanda

let current = 0;

const stream = new ReadableStream({
  pull(controller) {
    current += 1;
    controller.enqueue(current);

    if (current === 100) {
      controller.close();
    }
  }
});

pull() é chamado quando o consumidor sinaliza capacidade. Isso ajuda a respeitar backpressure.

Backpressure

Backpressure impede que o produtor gere dados muito mais rápido que o consumidor. O controller expõe desiredSize:

if (controller.desiredSize > 0) {
  controller.enqueue(nextChunk());
}

O valor é uma indicação, não uma garantia absoluta. O produtor deve parar quando não há demanda.

Queuing strategy

const stream = new ReadableStream(
  source,
  new CountQueuingStrategy({
    highWaterMark: 10
  })
);

O highWaterMark define quanta fila é desejada. Valores altos aumentam memória; valores baixos podem reduzir throughput.

ByteLengthQueuingStrategy

const strategy = new ByteLengthQueuingStrategy({
  highWaterMark: 64 * 1024
});

Essa estratégia mede o tamanho em bytes quando chunks possuem byteLength.

Cancelamento

const stream = new ReadableStream({
  cancel(reason) {
    console.log('Cancelado:', reason);
    closeExternalResource();
  }
});

Quando o consumidor cancela, libere arquivos, sockets, timers e listeners.

AbortSignal

Operações de piping podem aceitar sinal em versões compatíveis:

const controller = new AbortController();

await readable.pipeTo(writable, {
  signal: controller.signal
});

Veja AbortController no Node.js para timeouts e sinais compostos.

WritableStream

const writable = new WritableStream({
  write(chunk) {
    console.log('Recebido:', chunk);
  },
  close() {
    console.log('Concluído');
  },
  abort(reason) {
    console.error('Abortado:', reason);
  }
});

Usando writer

const writer = writable.getWriter();

try {
  await writer.write('A');
  await writer.write('B');
  await writer.close();
} finally {
  writer.releaseLock();
}

Aguardar write() permite respeitar a capacidade do destino.

writer.ready

await writer.ready;
await writer.write(chunk);

ready resolve quando a pressão diminui. Em um loop de produção, use essa Promise para não acumular dados.

TransformStream

const uppercase = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(String(chunk).toUpperCase());
  }
});

TransformStream possui lado gravável e lado legível.

Pipeline

await source
  .pipeThrough(uppercase)
  .pipeTo(destination);

Erros e cancelamentos são propagados conforme as opções do piping.

Transformação assíncrona

const transform = new TransformStream({
  async transform(chunk, controller) {
    const result = await processChunk(chunk);
    controller.enqueue(result);
  }
});

O processamento sequencial pode limitar concorrência. Se paralelizar, preserve ordem quando necessário e imponha limite de tarefas.

TextEncoderStream

const encoded = textReadable.pipeThrough(
  new TextEncoderStream()
);

Strings são convertidas em bytes UTF-8.

TextDecoderStream

const decoded = byteReadable.pipeThrough(
  new TextDecoderStream('utf-8')
);

O decoder preserva caracteres multibyte divididos entre chunks.

CompressionStream

Runtimes compatíveis podem oferecer CompressionStream:

const compressed = source.pipeThrough(
  new CompressionStream('gzip')
);

Verifique suporte e formatos. Para APIs nativas tradicionais, consulte Zlib no Node.js.

DecompressionStream

const decompressed = source.pipeThrough(
  new DecompressionStream('gzip')
);

Limite o tamanho descompactado para evitar ataques de compressão.

Fetch e Response

const response = await fetch(url);

if (!response.ok) {
  throw new Error(`HTTP ${response.status}`);
}

for await (const chunk of response.body) {
  consume(chunk);
}

O corpo de Response é uma ReadableStream. Isso permite processar respostas grandes sem carregar tudo em memória.

Response.text() e memória

response.text() acumula todo o conteúdo. Para arquivos grandes ou fluxos contínuos, consuma o body em chunks.

Criando uma Response

const response = new Response(stream, {
  headers: {
    'content-type': 'text/plain; charset=utf-8'
  }
});

Esse padrão é útil em runtimes que usam Request e Response como interfaces de servidor.

Tee

const [first, second] = stream.tee();

tee() divide o fluxo. Se um consumidor for lento, dados podem ser acumulados para ele. Use com cuidado em streams grandes.

pipeTo()

await readable.pipeTo(writable, {
  preventClose: false,
  preventAbort: false,
  preventCancel: false
});

Opções controlam propagação. Alterá-las sem entender pode deixar recursos abertos.

Erros no produtor

controller.error(new Error('Falha na origem'));

Depois de error, o stream não aceita novos chunks. Libere recursos externos.

Erros no consumidor

Um erro em WritableStream pode cancelar a origem durante piping. Registre o contexto e garanta cleanup idempotente.

Byte streams

const bytes = new ReadableStream({
  type: 'bytes',
  pull(controller) {
    const chunk = new Uint8Array([1, 2, 3]);
    controller.enqueue(chunk);
    controller.close();
  }
});

Byte streams oferecem otimizações e readers BYOB em cenários compatíveis.

BYOB reader

const reader = bytes.getReader({ mode: 'byob' });
const target = new Uint8Array(1024);
const result = await reader.read(target);

Bring Your Own Buffer reduz alocações, mas exige atenção a views e tamanhos.

Conversão de stream Node para Web

const { Readable } = require('node:stream');

const webReadable = Readable.toWeb(nodeReadable);

O suporte depende da versão e do tipo de stream.

Conversão de Web para Node

const nodeReadable = Readable.fromWeb(webReadable);

Teste tipos de chunks e tratamento de erros. Nem toda semântica é idêntica.

Writable adapters

O módulo stream também oferece adaptadores para Writable em versões modernas. Eles ajudam a integrar bibliotecas antigas e novas.

Objetos versus bytes

Streams tradicionais do Node.js possuem objectMode. Web Streams aceitam valores JavaScript, mas estratégias e consumidores precisam concordar com o formato. Documente se cada chunk é objeto, string ou Uint8Array.

Server-Sent Events

Web Streams podem produzir eventos contínuos:

const encoder = new TextEncoder();

const stream = new ReadableStream({
  start(controller) {
    const timer = setInterval(() => {
      controller.enqueue(
        encoder.encode(`data: ${Date.now()}\n\n`)
      );
    }, 1000);

    return () => clearInterval(timer);
  }
});

O cleanup precisa ser ligado ao cancelamento. Veja Server-Sent Events com Node.js.

Limites

Mesmo com streaming, cada chunk precisa de limite. Um parser que aceita linhas infinitas ou objetos gigantes continua vulnerável.

Timeouts

Defina prazo para início, inatividade e duração total. Cancelar o pipeline deve fechar a origem.

Observabilidade

Registre:

• bytes processados;
• chunks;
• duração;
• tempo bloqueado por backpressure;
• cancelamentos;
• erros;
• origem e destino sanitizados.

Testes

Cubra:

• stream vazio;
• múltiplos chunks;
• Unicode dividido;
• consumidor lento;
• cancelamento;
• erro de origem;
• erro de destino;
• transformação assíncrona;
• conversão Node/Web;
• limite de memória.

Erros comuns

• Não liberar lock: o stream fica indisponível.
• Ignorar desiredSize: a fila cresce.
• Usar tee em fluxo grande: um consumidor lento acumula dados.
• Carregar response.text(): todo o corpo vai para memória.
• Não tratar cancelamento: sockets e timers permanecem ativos.
• Misturar tipos de chunk: consumidores falham.
• Assumir semântica idêntica: adapters exigem testes.

Boas práticas

• Defina o tipo dos chunks.
• Respeite backpressure.
• Libere locks.
• Implemente cancelamento.
• Use limites de tamanho.
• Defina timeouts.
• Evite acumular todo o fluxo.
• Teste adapters.
• Monitore bytes e duração.
• Feche recursos em todos os caminhos.

Conclusão

A Web Streams API no Node.js oferece uma interface portável para fluxos legíveis, graváveis e transformações, integrando-se com fetch, Request e Response.

O principal benefício aparece quando a aplicação respeita backpressure, cancelamento e limites. Com adaptadores para streams tradicionais, é possível modernizar integrações gradualmente. Ao testar tipos de chunks, erros e consumidores lentos, Web Streams tornam o processamento de dados grandes mais eficiente e interoperável sem sacrificar a estabilidade do serviço.