Ga naar de inhoud

Streams

Bekijken als Markdown Bewerken op GitHub Documentatie op 5b53f69

Phasis implementeert de volledige WHATWG Streams Standard. Elke klasse, elke reader-modus, elke helper. Dezelfde code die in Node en moderne browsers werkt, werkt ook in Phasis.

De implementatie is wat fetch intern gebruikt — Response.body is een echte ReadableStream, Request accepteert streaming bodies en de tee() die u nodig hebt om een Response te klonen werkt.

Wat is geïmplementeerd

Klasse Status
ReadableStream Volledig
ReadableStreamDefaultController Volledig
ReadableByteStreamController Volledig
ReadableStreamDefaultReader Volledig
ReadableStreamBYOBReader Volledig
ReadableStreamBYOBRequest Volledig
WritableStream Volledig
WritableStreamDefaultController Volledig
WritableStreamDefaultWriter Volledig
TransformStream Volledig
TransformStreamDefaultController Volledig
ByteLengthQueuingStrategy Volledig
CountQueuingStrategy Volledig

En de methoden die u vaak gebruikt: pipeTo, pipeThrough, tee, getReader, asynchrone iteratie, Symbol.asyncIterator, BYOB-lezen met { min }.

Produceren

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

Consumeren

Drie manieren om te lezen:

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

Transformeren

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

Bytestreams (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 } wordt ondersteund — read(view, { min: 2 }) wordt pas opgelost wanneer er minstens 2 elementen beschikbaar zijn.

Backpressure

desiredSize en ready werken zoals de specificatie beschrijft. Gebruik een wachtrijstrategie om de producent een terugkanaal te geven:

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

Met 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 voor caching

Veelgebruikt patroon: lees de body tweemaal.

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

Response.clone() doet dit voor u en retourneert een nieuwe Response.

Kanttekeningen

  • Alleen synchrone producenten. Een pull() die blokkerende PHP-I/O uitvoert, blokkeert het script. Gebruik de fetch-transporthook voor HTTP; lees bestanden bij voorkeur vooraf in het geheugen. setTimeout / setInterval zijn beschikbaar (zie Eventloop), maar de engine gebruikt standaard virtuele tijd — een setTimeout(cb, 1000) binnen een pull() slaapt niet PHP, maar zet de klok van de eventloop vooruit. Voor realtime timing moet de host tickEventLoop() aansturen op basis van een wandklok.

Zie ook