Gå til indhold

Streams

Vis som Markdown Rediger på GitHub Dokumentation hos 5b53f69

Phasis implementerer hele WHATWG Streams Standard. Hver klasse, hver læsertilstand og hver hjælper. Den samme kode, der fungerer i Node og moderne browsere, fungerer i Phasis.

Implementeringen er den, fetch selv bruger internt — Response.body er en ægte ReadableStream, Request accepterer streaming bodies, og tee(), du skal bruge til at klone et Response, fungerer.

Det, der er implementeret

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

Samt de metoder, du typisk bruger: pipeTo, pipeThrough, tee, getReader, asynkron iteration, Symbol.asyncIterator, BYOB-læsning med { min }.

Produktion

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

Forbrug

Tre måder at læse på:

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

Transformation

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 } understøttes — read(view, { min: 2 }) løses først, når der er mindst 2 elementer til rådighed.

Backpressure

desiredSize og ready fungerer som beskrevet i specifikationen. Brug en køstrategi til at give producenten en tilbagemeldingskanal:

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

Med 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 til cachelagring

Almindeligt mønster: læs body to gange.

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

Response.clone() gør det for dig og returnerer en ny Response.

Forbehold

  • Kun synkrone producenter. En pull(), der udfører blokerende PHP I/O, blokerer scriptet. Brug fetch-transport-hooket til HTTP; for filer er det bedst at læse dem ind i hukommelsen først. setTimeout / setInterval er tilgængelige (se Hændelsesløkke), men motoren bruger som standard virtuel tid — en setTimeout(cb, 1000) i en pull() sover ikke PHP, men flytter løkkens ur frem. Realtidstiming kræver, at værten driver tickEventLoop() efter et vægur.

Se også