Saltar al contenido

Streams

Ver como Markdown Editar en GitHub Documentación de 5b53f69

Phasis implementa el estándar WHATWG Streams completo. Todas las clases, todos los modos de lector y todos los helpers. El mismo código que funciona en Node y en los navegadores modernos funciona en Phasis.

Esta es la implementación que fetch usa internamente: Response.body es un ReadableStream real, Request acepta cuerpos en streaming y funciona tee(), que necesitas para clonar una Response.

Qué incluye

Clase Estado
ReadableStream Completa
ReadableStreamDefaultController Completa
ReadableByteStreamController Completa
ReadableStreamDefaultReader Completa
ReadableStreamBYOBReader Completa
ReadableStreamBYOBRequest Completa
WritableStream Completa
WritableStreamDefaultController Completa
WritableStreamDefaultWriter Completa
TransformStream Completa
TransformStreamDefaultController Completa
ByteLengthQueuingStrategy Completa
CountQueuingStrategy Completa

Además, incluye los métodos que más se usan: pipeTo, pipeThrough, tee, getReader, iteración asíncrona, Symbol.asyncIterator y lectura BYOB con { min }.

Producción

JavaScript
// Pull-based — only generates values when read
const counter = new ReadableStream({
  start(controller) { this.n = 0; },
  pull(controller) {
    if (this.n >= 5) { controller.close(); return; }
    controller.enqueue(this.n++);
  },
});

for await (const v of counter) console.log(v);  // 0,1,2,3,4
JavaScript
// Push-based — start enqueues everything eagerly
const small = new ReadableStream({
  start(c) {
    [10, 20, 30].forEach(n => c.enqueue(n));
    c.close();
  },
});

Consumo

Tres formas de leer:

JavaScript
// 1. Reader
const reader = stream.getReader();
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  console.log(value);
}
reader.releaseLock();

// 2. Async iteration (preferred)
for await (const chunk of stream) {
  console.log(chunk);
}

// 3. tee + parallel pipes
const [a, b] = stream.tee();
await Promise.all([a.pipeTo(sinkA), b.pipeTo(sinkB)]);

Transformación

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

const result = new ReadableStream({
  start(c) { ['hi', 'there'].forEach(s => c.enqueue(s)); c.close(); }
}).pipeThrough(upper);

for await (const v of result) console.log(v);  // HI, THERE

Streams de bytes (BYOB)

JavaScript
const bytes = new ReadableStream({
  type: 'bytes',
  start(controller) {
    controller.enqueue(new Uint8Array([1, 2, 3, 4, 5, 6]));
    controller.close();
  },
});

const reader = bytes.getReader({ mode: 'byob' });
const buf = new Uint8Array(4);
const { value, done } = await reader.read(buf);
// value is a Uint8Array of length 4, sharing buffer with `buf`

{ min: N } es compatible: read(view, { min: 2 }) solo se resuelve cuando hay al menos 2 elementos disponibles.

Control de contrapresión

desiredSize y ready funcionan según lo descrito en la especificación. Usa una estrategia de encolado para proporcionar un canal de retorno al productor:

JavaScript
const stream = new ReadableStream({
  pull(controller) {
    if (controller.desiredSize <= 0) return;  // back off
    controller.enqueue(nextChunk());
  },
}, new CountQueuingStrategy({ highWaterMark: 4 }));

Con fetch

JavaScript
// Response body is a ReadableStream
const res = await fetch('https://example.com/big.txt');
let total = 0;
for await (const chunk of res.body) {
  total += chunk.byteLength;
}

// Request bodies can stream too
const upload = new ReadableStream({
  start(c) { c.enqueue(new TextEncoder().encode('hello')); c.close(); }
});
await fetch('https://example.com/upload', { method: 'POST', body: upload });

tee para el almacenamiento en caché

Patrón habitual: leer el cuerpo dos veces.

JavaScript
const res = await fetch(url);
const [forCache, forUse] = res.body.tee();
await Promise.all([
  forCache.pipeTo(cacheWritable),
  parseJson(forUse),
]);

Response.clone() lo hace por ti y devuelve una Response nueva.

Advertencias

  • Productores solo síncronos. Una pull() que realice operaciones PHP bloqueantes bloqueará el script. Usa el hook de transporte fetch para HTTP; con archivos, es mejor leerlos antes y cargarlos en memoria. setTimeout / setInterval están disponibles (consulta Bucle de eventos), pero el motor usa tiempo virtual de forma predeterminada: un setTimeout(cb, 1000) dentro de un pull() no duerme PHP, sino que adelanta el reloj del bucle. Para marcar el ritmo en tiempo real, el anfitrión debe ejecutar tickEventLoop() usando un reloj de pared.

Véase también