Vai al contenuto

Flussi

Visualizza come Markdown Modifica su GitHub Documentazione di 5b53f69

Phasis implementa l'intero standard WHATWG Streams. Ogni classe, ogni modalità di lettura e ogni helper. Lo stesso codice che funziona in Node e nei browser moderni funziona anche in Phasis.

È l'implementazione che fetch usa internamente: Response.body è un vero ReadableStream, Request accetta corpi in streaming e funziona il metodo tee() necessario per clonare una Response.

Funzionalità incluse

Classe Stato
ReadableStream Completo
ReadableStreamDefaultController Completo
ReadableByteStreamController Completo
ReadableStreamDefaultReader Completo
ReadableStreamBYOBReader Completo
ReadableStreamBYOBRequest Completo
WritableStream Completo
WritableStreamDefaultController Completo
WritableStreamDefaultWriter Completo
TransformStream Completo
TransformStreamDefaultController Completo
ByteLengthQueuingStrategy Completo
CountQueuingStrategy Completo

Inoltre, i metodi che si usano più spesso: pipeTo, pipeThrough, tee, getReader, iterazione asincrona, Symbol.asyncIterator, lettura BYOB con { min }.

Produzione

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

Tre modi per leggere:

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

Trasformazione

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

Flussi di byte (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 } è supportato — read(view, { min: 2 }) si risolve solo quando sono disponibili almeno 2 elementi.

Controllo della pressione

desiredSize e ready funzionano come descritto nella specifica. Usate una strategia di accodamento per fornire al produttore un canale di ritorno:

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 per la cache

Schema comune: leggere il corpo due volte.

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

Response.clone() lo fa per voi e restituisce una nuova Response.

Avvertenze

  • Produttori solo sincroni. Un pull() che esegue operazioni I/O bloccanti PHP blocca lo script. Per HTTP, usate l'hook di trasporto fetch; per i file, preferite caricarli in memoria in anticipo. Sono disponibili setTimeout / setInterval (consultate Ciclo degli eventi), ma per impostazione predefinita il motore usa il tempo virtuale: un setTimeout(cb, 1000) dentro un pull() non attende PHP, ma fa avanzare l'orologio del ciclo. Per scandire il tempo reale, l'host deve far avanzare tickEventLoop() in base all'orologio reale.

Vedi anche