Aller au contenu

Flux

Afficher au format Markdown Modifier sur GitHub Documentation de 5b53f69

Phasis implémente l'intégralité de la norme WHATWG Streams. Chaque classe, chaque mode de lecteur, chaque assistant. Le même code qui fonctionne dans Node et les navigateurs modernes fonctionne dans Phasis.

C'est l'implémentation que fetch utilise en interne : Response.body est un véritable ReadableStream, Request accepte les corps en flux continu, et le tee() dont vous avez besoin pour cloner une Response fonctionne.

Fonctionnalités disponibles

Classe État
ReadableStream Complet
ReadableStreamDefaultController Complet
ReadableByteStreamController Complet
ReadableStreamDefaultReader Complet
ReadableStreamBYOBReader Complet
ReadableStreamBYOBRequest Complet
WritableStream Complet
WritableStreamDefaultController Complet
WritableStreamDefaultWriter Complet
TransformStream Complet
TransformStreamDefaultController Complet
ByteLengthQueuingStrategy Complet
CountQueuingStrategy Complet

Et aussi les méthodes courantes : pipeTo, pipeThrough, tee, getReader, l'itération asynchrone, Symbol.asyncIterator, la lecture BYOB avec { min }.

Produire

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();
  },
});

Consommer

Trois façons de lire :

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

Transformer

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

Flux d'octets (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 } est pris en charge — read(view, { min: 2 }) ne se résout que lorsqu'au moins 2 éléments sont disponibles.

Contre-pression

desiredSize et ready fonctionnent comme le décrit la spécification. Utilisez une stratégie de mise en file d'attente pour renvoyer des informations au producteur :

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

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

Duplication pour la mise en cache

Modèle courant : lire le corps deux fois.

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

Response.clone() s'en charge et renvoie un nouvel objet Response.

Réserves

  • Producteurs synchrones uniquement. Un pull() qui effectue des E/S PHP bloquantes bloquera le script. Utilisez le hook de transport fetch pour HTTP ; pour les fichiers, privilégiez leur chargement complet en mémoire au préalable. setTimeout / setInterval sont disponibles (voir Boucle d'événements), mais le moteur utilise par défaut le temps virtuel : un setTimeout(cb, 1000) dans un pull() ne fait pas attendre PHP, il avance l'horloge de la boucle. Pour cadencer en temps réel, l'hôte doit piloter tickEventLoop() en fonction de l'horloge réelle.

Voir aussi