Transform & Duplex

Streams Transform ficam no meio de pipelines para analisar, comprimir ou filtrar - Streams Duplex modelam canais bidirecionais como sockets com lados de leitura e escrita independentes.

Receita

import { Transform, PassThrough } from 'node:stream';
import { pipeline } from 'node:stream/promises';
 
const tap = new PassThrough();
tap.on('data', (chunk) => metrics.bytes(chunk.length));
 
await pipeline(source, tap, destination);

Quando usar isso:

• Parse NDJSON → pipeline em modo objeto
• Compressão gzip no caminho de upload
• Duplicar um download para o cliente e um scanner de vírus
• Divisão de delimitador de linha em streams TCP

Exemplo Funcional

import { Transform, PassThrough, Duplex } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { Readable } from 'node:stream';
 
const splitLines = new Transform({
  transform(chunk, _enc, cb) {
    const parts = String(chunk).split('\n');
    const tail = parts.pop() ?? '';
    for (const line of parts) if (line) this.push(line);
    cb(null, tail); // armazena linha incompleta no estado interno passando como chunk para o próximo
  },
  flush(cb) {
    cb();
  },
});
 
function tee<T extends NodeJS.ReadableStream>(source: T): [T, PassThrough] {
  const branch = new PassThrough();
  source.pipe(branch);
  return [source, branch];
}
 
// Duplex conceitual - net.Socket estende Duplex
const socketLike = new Duplex({
  read() { /* puxa do recurso subjacente */ },
  write(chunk, _enc, cb) { cb(); /* envia para o par */ },
});

import { createGzip } from 'node:zlib';
 
await pipeline(
  Readable.from(['{"a":1}\n', '{"b":2}\n']),
  splitLines,
  createGzip(),
  process.stdout,
);

O que isso demonstra:

• Transform pode armazenar tokens parciais entre chunks (divisão de linha)
• PassThrough é um Transform de identidade - ideal para canais laterais de tap/tee
• zlib.createGzip() é um Transform internamente
• read/write do Duplex são independentes - diferente do fluxo ligado do Transform

Mergulho Profundo

Como Funciona

• Transform implementa _transform por chunk e _flush opcional no final.
• Duplex implementa _read e _write separadamente - sockets TCP, TLS.
• PassThrough - retransmissão de cópia zero com as mesmas regras de backpressure.
• Tee - múltiplos consumidores de uma fonte exigem backpressure cuidadosa (pausa todos se um for lento).

Escolha do Tipo de Stream

Notas de TypeScript

import { Transform, type TransformCallback } from 'node:stream';
 
function createJsonParser(): Transform {
  let buffer = '';
  return new Transform({
    objectMode: true,
    transform(chunk: Buffer, _enc: BufferEncoding, cb: TransformCallback) {
      buffer += chunk.toString('utf8');
      const lines = buffer.split('\n');
      buffer = lines.pop() ?? '';
      try {
        for (const line of lines) if (line) this.push(JSON.parse(line));
        cb();
      } catch (err) {
        cb(err as Error);
      }
    },
  });
}

Armadilhas

• Tee sem lidar com backpressure de branch lento - um consumidor lento paralisa todos. Correção: usar filas limitadas ou tópicos de ingestão separados.
• Transform não chamando cb em caminhos de erro - pipeline travado. Correção: sempre cb(err) uma vez.
• Misturar estado de flush incorretamente no parser de linha - linha final perdida. Correção: _flush emite o restante.
• Assumir que read/write do Duplex compartilham buffer - eles não compartilham a menos que você os conecte.
• Transform em modo objeto após gzip de bytes - a ordem importa - gzip espera bytes.

Alternativas

FAQs

Relacionados

• stream/promises.pipeline - conecta transforms com segurança
• Padrões Readable & Writable - modos
• Protocolos Binários - parsers de framing
• Backpressure - lentidão do tee

Versões da Stack: 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.