Históricamente, el ecosistema de Node.js dependió exclusivamente de su arquitectura nativa de streams (node:stream), construida sobre EventEmitter, callbacks y buffers de memoria binaria. En paralelo, el consorcio WHATWG estandarizó la Web Streams API para navegadores y entornos Edge. El estándar moderno de la web se cimenta sobre contratos Promise, arrays tipados Uint8Array, bloqueos exclusivos de lectura (Reader Locks) y pipelines funcionales pipeThrough.
En el stack fullstack contemporáneo (Node.js 18-24+, Next.js 15+, React 19, Cloudflare Workers y Vercel Serverless), la Web Streams API se ha consolidado como la base universal de transferencia de datos. La API global fetch(), los interfaces nativos Request y Response, el streaming de React Server Components (protocolo RSC Flight) y los Route Handlers de Next.js utilizan el estándar WHATWG Streams por defecto.
Comprender la arquitectura de Web Streams y dominar su integración bidireccional con los streams tradicionales de Node.js (node:stream) constituye una competencia indispensable para construir arquitecturas de backend de alto rendimiento y aplicaciones SSR modernas.
La especificación de Web Streams API define tres componentes cardinales:
ReadableStream: Fuente de datos gestionada mediante un controlador interno:
ReadableStreamDefaultController: Controlador de propósito general que entrega chunks de datos arbitrarios.ReadableByteStreamController (BYOB — Bring Your Own Buffer): Controlador optimizado para E/S binaria de bajo nivel. Permite al consumidor proporcionar su propio buffer preasignado, eliminando copias superfluas de memoria en V8.WritableStream: Destino receptor de datos gestionado mediante WritableStreamDefaultController.TransformStream: Pareja de streams acoplados compuesta por un WritableStream de entrada (stream.writable) y un ReadableStream de salida (stream.readable), diseñada para transformar o recodificar chunks sobre la marcha.A diferencia de los streams clásicos de Node.js, donde pueden registrarse múltiples listeners 'data' de forma simultánea, los Web Streams imponen un modelo estricto de consumidor único mediante bloqueos:
const reader = readableStream.getReader();
console.log(readableStream.locked); // trueEn cuanto se invoca getReader(), el stream pasa al estado bloqueado (locked === true). Ningún otro componente puede interactuar con el stream ni llamar a pipeTo() hasta que el reader actual invoque reader.releaseLock() o cierre la lectura. Si se requiere enviar un mismo stream a dos destinos independientes, debe bifurcarse explícitamente mediante readableStream.tee().
La contrapresión en Web Streams se gobierna mediante estrategias de cola (Queuing Strategies):
CountQueuingStrategy({ highWaterMark: N }): Mide la ocupación del buffer en cantidad de chunks discretos.ByteLengthQueuingStrategy({ highWaterMark: N }): Calcula el volumen total de bytes acumulados según la propiedad chunk.byteLength.Cuando el tamaño de los elementos en cola alcanza highWaterMark, la propiedad controller.desiredSize desciende a cero o menos (desiredSize <= 0). Esta señal indica al productor que debe pausar la generación de chunks.
Para solventar la disparidad entre ecosistemas sin coste de copia de memoria, el módulo node:stream incorpora adaptadores bidireccionales zero-copy:
Readable.toWeb(nodeReadable): Transforma un stream clásico de Node.js en un ReadableStream de WHATWG.Readable.fromWeb(webReadableStream): Convierte un ReadableStream estándar en un Readable clásico de node:stream.Writable.toWeb(nodeWritable) y Writable.fromWeb(webWritableStream): Adaptadores homólogos para streams de escritura.Un pipeline de Web Streams conforma una cadena funcional ininterrumpida con un control estricto de bloqueos y propagación reactiva de contrapresión:
[ ReadableStream ] (locked: true)
|
| (pipeThrough)
v
+-----------------------+
| TransformStream |
| [writable] -> [readable]
+-----------------------+
|
| (pipeThrough)
v
+------------------------------------+
| CompressionStream('gzip') |
| [writable: raw] -> [readable: gzip]|
+------------------------------------+
|
| (pipeTo)
v
[ WritableStream / HTTP Response ]new Response(readableStream).<Suspense>.CompressionStream('gzip') de forma idéntica en Node.js, Deno y Edge Workers.El siguiente generador de eventos SSE supervisa la propiedad desiredSize para respetar la contrapresión:
export function createIncrementalStream(maxItems: number): ReadableStream<Uint8Array> {
let current = 0;
const encoder = new TextEncoder();
return new ReadableStream<Uint8Array>({
start(controller) {
console.log('Stream inicializado');
},
// pull() se ejecuta automáticamente cuando el buffer downstream está listo para recibir datos
async pull(controller) {
while ((controller.desiredSize ?? 0) > 0 && current < maxItems) {
current++;
const chunk = encoder.encode(`data: {"id": ${current}, "timestamp": ${Date.now()}}\n\n`);
controller.enqueue(chunk);
if (current % 100 === 0) {
await new Promise((resolve) => setTimeout(resolve, 10));
}
}
if (current >= maxItems) {
controller.close();
}
},
cancel(reason) {
console.warn('Stream cancelado por el cliente:', reason);
},
}, new ByteLengthQueuingStrategy({ highWaterMark: 64 * 1024 }));
}Los clientes de bases de datos y SDKs de nube habitualmente devuelven instancias de node:stream. Para emplearlas en Route Handlers de Next.js se requiere el puente Readable.toWeb():
import { Readable, Writable } from 'node:stream';
import { createReadStream } from 'node:fs';
// 1. Conversión de stream de lectura de Node.js a WHATWG ReadableStream
const nodeReadable = createReadStream('large-dataset.csv');
const webReadableStream: ReadableStream<Uint8Array> = Readable.toWeb(nodeReadable);
// 2. Consumo de Web Stream con bloqueo exclusivo de reader
async function consumeWebStream(stream: ReadableStream<Uint8Array>): Promise<number> {
const reader = stream.getReader();
let totalBytes = 0;
try {
while (true) {
const { value, done } = await reader.read();
if (done) break;
totalBytes += value.byteLength;
}
} finally {
// Es crítico liberar el bloqueo en el bloque finally
reader.releaseLock();
}
return totalBytes;
}
// 3. Conversión inversa: WHATWG ReadableStream -> Node.js Readable
const convertedBackToNode: Readable = Readable.fromWeb(webReadableStream);La clase estándar CompressionStream permite componer pipelines de compresión Gzip sin dependencias externas de npm:
import { NextRequest, NextResponse } from 'next/server';
import { Readable } from 'node:stream';
import { createReadStream } from 'node:fs';
import { join } from 'node:path';
export async function GET(req: NextRequest): Promise<Response> {
const filePath = join(process.cwd(), 'data', 'audit-records.json');
// 1. Obtenemos el stream de lectura nativo de Node.js
const fileReadStream = createReadStream(filePath);
// 2. Lo adaptamos a WHATWG ReadableStream
const sourceWebStream = Readable.toWeb(fileReadStream);
// 3. Lo encadenamos con compresión GZIP nativa
const compressedStream = sourceWebStream.pipeThrough(
new CompressionStream('gzip')
);
// 4. Devolvemos la respuesta HTTP con transferencia por bloques
return new Response(compressedStream, {
status: 200,
headers: {
'Content-Type': 'application/json',
'Content-Encoding': 'gzip',
'Transfer-Encoding': 'chunked',
'Cache-Control': 'no-cache, no-store',
},
});
}Lectura del archivo mediante fs.createReadStream con chunks de 64 KB.
Envoltura del stream de Node.js en un ReadableStream sin copiar memoria.
Compresión dinámica al vuelo mediante CompressionStream('gzip').
Creación del objeto Response estándar con cabecera Transfer-Encoding: chunked.
El navegador descomprime y procesa los datos progresivamente sobre HTTP/2.
Un módulo isomórfico para construir pipelines fiables de Web Streams con soporte para Server-Sent Events, compresión y cancelación:
export interface StreamPipelineOptions {
compression?: 'gzip' | 'deflate' | 'none';
signal?: AbortSignal;
highWaterMarkBytes?: number;
onChunkSent?: (bytes: number) => void;
}
export class WebStreamPipelineHelper {
private static readonly DEFAULT_HIGH_WATER_MARK = 64 * 1024; // 64 KB
/**
* Envuelve un generador asíncrono en un ReadableStream optimizado.
*/
public static fromAsyncGenerator<T>(
generator: AsyncGenerator<T, void, unknown>,
serializer: (item: T) => Uint8Array,
options: StreamPipelineOptions = {}
): ReadableStream<Uint8Array> {
const highWaterMark = options.highWaterMarkBytes ?? this.DEFAULT_HIGH_WATER_MARK;
const baseStream = new ReadableStream<Uint8Array>(
{
async pull(controller) {
try {
const { value, done } = await generator.next();
if (done) {
controller.close();
return;
}
const binaryChunk = serializer(value);
controller.enqueue(binaryChunk);
if (options.onChunkSent) {
options.onChunkSent(binaryChunk.byteLength);
}
} catch (err) {
controller.error(err);
}
},
cancel(reason) {
console.warn('Pipeline cancelado por el consumidor:', reason);
},
},
new ByteLengthQueuingStrategy({ highWaterMark })
);
return this.applyCompression(baseStream, options.compression);
}
/**
* Aplica compresión al pipeline si se solicita.
*/
public static applyCompression(
stream: ReadableStream<Uint8Array>,
compression: 'gzip' | 'deflate' | 'none' = 'none'
): ReadableStream<Uint8Array> {
if (compression === 'none') {
return stream;
}
const compressor = new CompressionStream(compression);
return stream.pipeThrough(compressor);
}
/**
* Genera cabeceras HTTP de producción para respuestas en streaming.
*/
public static createStreamHeaders(
contentType: string = 'text/event-stream',
compression: 'gzip' | 'deflate' | 'none' = 'none'
): Headers {
const headers = new Headers({
'Content-Type': contentType,
'Cache-Control': 'no-cache, no-transform',
'Connection': 'keep-alive',
'X-Content-Type-Options': 'nosniff',
});
if (compression !== 'none') {
headers.set('Content-Encoding', compression);
}
return headers;
}
}Llamar a stream.getReader() en un módulo y luego intentar pasar el mismo stream a new Response(stream) o invocar stream.pipeThrough(). El entorno arroja un error crítico TypeError: Cannot get a reader for a locked stream.
Libera siempre el bloqueo mediante reader.releaseLock() antes de reutilizar el stream, o bifúrcalo previamente con const [branchA, branchB] = stream.tee().
Cuando la fuente de datos se agota, olvidar invocar controller.close() hace que el cliente de red espere indefinidamente bytes adicionales. La conexión queda suspendida hasta el timeout, bloqueando recursos del servidor.
Asegúrate siempre de llamar a controller.close() al finalizar los datos y de propagar los fallos con controller.error(err).
Pasar fs.createReadStream() directamente al constructor new Response(nodeStream) en Next.js o Node.js. El constructor espera un ReadableStream (WHATWG) o un AsyncIterable. Esto desencadena errores de tipado o respuestas vacías con pérdida de datos.
Envuelve siempre los streams de Node.js con Readable.toWeb(nodeStream).
Si un usuario cierra la pestaña del navegador o detiene la descarga, no supervisar la señal de cancelación provoca que el servidor continúe procesando chunks y ejecutando consultas en bases de datos.
Transfiere siempre req.signal desde NextRequest a los generadores de datos y suscríbete al evento abort.
El cliente envía la petición HTTP al Route Handler solicitando Accept-Encoding: gzip.
Creación de ReadableStream supervisando la capacidad de la cola con desiredSize.
Paso de los datos a través de CompressionStream('gzip') mediante pipeThrough().
Transmisión de los chunks comprimidos con la cabecera Transfer-Encoding: chunked.
Invocación de controller.close() al finalizar los datos, liberando recursos del sistema.