Zum Inhalt springen

Streams

Als Markdown anzeigen Auf GitHub bearbeiten Dokumentation zu 5b53f69

Phasis implementiert den vollständigen WHATWG-Streams-Standard. Jede Klasse, jeder Reader-Modus und jeder Helper. Derselbe Code, der in Node und modernen Browsern funktioniert, läuft auch in Phasis.

Die Implementierung verwendet fetch intern: Response.body ist ein echter ReadableStream, Request akzeptiert Streaming-Bodies, und auch der tee(), den du zum Klonen einer Response brauchst, funktioniert.

Enthalten

Klasse Status
ReadableStream Vollständig
ReadableStreamDefaultController Vollständig
ReadableByteStreamController Vollständig
ReadableStreamDefaultReader Vollständig
ReadableStreamBYOBReader Vollständig
ReadableStreamBYOBRequest Vollständig
WritableStream Vollständig
WritableStreamDefaultController Vollständig
WritableStreamDefaultWriter Vollständig
TransformStream Vollständig
TransformStreamDefaultController Vollständig
ByteLengthQueuingStrategy Vollständig
CountQueuingStrategy Vollständig

Außerdem die Methoden, die du regelmäßig brauchst: pipeTo, pipeThrough, tee, getReader, asynchrone Iteration, Symbol.asyncIterator, BYOB-Lesen mit { min }.

Daten erzeugen

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

Daten lesen

Drei Möglichkeiten zum Lesen:

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

Daten transformieren

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

Byte-Streams (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 } wird unterstützt — read(view, { min: 2 }) wird erst aufgelöst, wenn mindestens 2 Elemente verfügbar sind.

Rückstau

desiredSize und ready funktionieren wie in der Spezifikation beschrieben. Verwende eine Warteschlangenstrategie, um dem Producer einen Rückkanal bereitzustellen:

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

Mit 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 zum Cachen

Gängiges Muster: den Body zweimal lesen.

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

Response.clone() übernimmt das für dich und gibt einen neuen Response zurück.

Einschränkungen

  • Nur synchrone Producer. Ein pull(), das blockierende PHP-E/A ausführt, blockiert das Skript. Verwende für HTTP den fetch-Transport-Hook; bei Dateien solltest du sie vorher in den Speicher einlesen. setTimeout / setInterval sind verfügbar (siehe Event loop), aber die Engine verwendet standardmäßig virtuelle Zeit — ein setTimeout(cb, 1000) innerhalb eines pull() schläft nicht PHP, sondern stellt die Loop-Uhr vor. Für Echtzeit-Taktung muss der Host tickEventLoop() anhand einer Echtzeituhr steuern.

Siehe auch