Ugrás a tartalomhoz

Adatfolyamok

Megtekintés Markdownként Szerkesztés itt: GitHub Dokumentáció itt: 5b53f69

A(z) Phasis megvalósítja a teljes WHATWG Streams Standard szabványt. Minden osztály, minden olvasómód és minden segédfüggvény elérhető. Ugyanaz a kód, amely Node-ban és a modern böngészőkben működik, a(z) Phasis környezetben is működik.

A(z) fetch belsőleg is ezt a megvalósítást használja — a(z) Response.body valódi ReadableStream, a(z) Request adatfolyamos törzseket fogad, és működik a(z) tee() is, amellyel egy Response klónozható.

Elérhető funkciók

Osztály Állapot
ReadableStream Teljes
ReadableStreamDefaultController Teljes
ReadableByteStreamController Teljes
ReadableStreamDefaultReader Teljes
ReadableStreamBYOBReader Teljes
ReadableStreamBYOBRequest Teljes
WritableStream Teljes
WritableStreamDefaultController Teljes
WritableStreamDefaultWriter Teljes
TransformStream Teljes
TransformStreamDefaultController Teljes
ByteLengthQueuingStrategy Teljes
CountQueuingStrategy Teljes

És a gyakran használt metódusok: pipeTo, pipeThrough, tee, getReader, aszinkron iteráció, Symbol.asyncIterator, BYOB-olvasás a(z) { min } használatával.

Előállítás

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

Fogyasztás

Háromféleképpen olvashatja be:

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

Átalakítás

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

Bájtfolyamok (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`

A(z) { min: N } támogatott — a(z) read(view, { min: 2 }) csak akkor oldódik fel, ha legalább 2 elem elérhető.

Visszanyomás

A(z) desiredSize és ready a specifikációban leírtak szerint működik. Használjon sorba állítási stratégiát, hogy visszajelzési csatornát biztosítson az előállítónak:

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

Fetch használata

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

Kettéágaztatás gyorsítótárazáshoz

Gyakori minta: olvassa be kétszer a törzset.

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

A(z) Response.clone() ezt elvégzi Ön helyett, és egy új Response objektumot ad vissza.

Tudnivalók

  • Csak szinkron adatforrások. A pull(), amely blokkoló PHP I/O-t végez, blokkolja a szkriptet. HTTP esetén használja a fetch átviteli hookot; fájloknál inkább olvassa be előre az adatokat a memóriába. A setTimeout / setInterval is rendelkezésre áll (lásd: Eseményhurok), de a motor alapértelmezés szerint virtuális időt használ — egy setTimeout(cb, 1000) egy pull() belsejében nem alszik PHP, hanem előrelépteti a hurok óráját. A valós idejű ütemezéshez a gazdagépnek falióra szerint kell futtatnia a tickEventLoop() műveletet.

Lásd még