Transform y Duplex

Los streams Transform se sitúan en medio de los pipelines para analizar, comprimir o filtrar. Los streams Duplex modelan canales bidireccionales como los sockets con lados de lectura y escritura independientes.

Receta

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

Cuándo usarlo:

• Pipeline de análisis NDJSON a modo de objeto
• Compresión gzip en la ruta de subida
• Duplicar una descarga para el cliente y el escáner de virus
• División de delimitadores de línea en streams TCP

Ejemplo de funcionamiento

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); // almacena la línea incompleta en el estado interno pasándola como chunk a la siguiente
  },
  flush(cb) {
    cb();
  },
});
 
function tee<T extends NodeJS.ReadableStream>(source: T): [T, PassThrough] {
  const branch = new PassThrough();
  source.pipe(branch);
  return [source, branch];
}
 
// Duplex conceptual - net.Socket extiende Duplex
const socketLike = new Duplex({
  read() { /* extrae del recurso subyacente */ },
  write(chunk, _enc, cb) { cb(); /* envía al par */ },
});

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

Lo que esto demuestra:

• Transform puede almacenar tokens parciales en búfer a través de chunks (división de líneas)
• PassThrough es un Transform de identidad, ideal para canales laterales tap/tee
• zlib.createGzip() es un Transform subyacente
• Las operaciones read/write de Duplex son independientes, a diferencia del flujo vinculado de Transform

Análisis profundo

Cómo funciona

• Transform implementa _transform por chunk y un _flush opcional al final.
• Duplex implementa _read y _write por separado - sockets TCP, TLS.
• PassThrough - retransmisión casi sin copia con las mismas reglas de contrapresión.
• Tee - múltiples consumidores en una fuente requieren una contrapresión cuidadosa (pausar todos si uno es lento).

Selección de 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);
      }
    },
  });
}

Errores comunes

• Tee sin manejar la contrapresión de la rama lenta - un consumidor lento detiene a todos. Solución: usa colas acotadas o temas de ingesta separados.
• Transform no llama a cb en rutas de error - pipeline colgado. Solución: siempre cb(err) una vez.
• Mezclar el estado de vaciado incorrectamente en el analizador de líneas - línea final perdida. Solución: _flush empuja el resto.
• Asumir que la lectura/escritura de Duplex comparten búfer - no lo hacen a menos que los conectes.
• Transform en modo objeto después de gzip de bytes - el orden importa - gzip espera bytes.

Alternativas

Preguntas frecuentes

Relacionado

• stream/promises.pipeline - conecta transforms de forma segura
• Patrones Readable y Writable - modos
• Protocolos binarios - analizadores de tramas
• Contrapresión - ralentizaciones de tee

Versiones de la pila: Esta página fue escrita para Node.js 24.18.0 (LTS activa), npm 10+, TypeScript 5.6+, Express 5, Fastify 5 y NestJS 11.