Siirry sisältöön

Virrat

Näytä Markdown-muodossa Muokkaa palvelussa GitHub Dokumentaatio: 5b53f69

Phasis toteuttaa koko WHATWG Streams -standardin. Jokaisen luokan, jokaisen lukijatilanteen ja jokaisen apufunktion. Sama koodi, joka toimii Nodessa ja moderneissa selaimissa, toimii myös kohteessa Phasis.

Toteutus on se, jota fetch käyttää sisäisesti — Response.body on todellinen ReadableStream, Request hyväksyy virtaavat rungot, ja myös Response-olion kloonaamiseen tarvittava tee() toimii.

Julkaistu sisältö

Luokka Tila
ReadableStream Täydellinen
ReadableStreamDefaultController Täydellinen
ReadableByteStreamController Täydellinen
ReadableStreamDefaultReader Täydellinen
ReadableStreamBYOBReader Täydellinen
ReadableStreamBYOBRequest Täydellinen
WritableStream Täydellinen
WritableStreamDefaultController Täydellinen
WritableStreamDefaultWriter Täydellinen
TransformStream Täydellinen
TransformStreamDefaultController Täydellinen
ByteLengthQueuingStrategy Täydellinen
CountQueuingStrategy Täydellinen

Lisäksi käytetyimmät metodit: pipeTo, pipeThrough, tee, getReader, asynkroninen iteraatio, Symbol.asyncIterator, BYOB-luku kohteella { min }.

Tuottaminen

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

Kuluttaminen

Kolme tapaa lukea:

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

Muuntaminen

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

Tavuvirrat (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 } on tuettu — read(view, { min: 2 }) ratkeaa vain, kun vähintään 2 elementtiä on käytettävissä.

Vastapaine

desiredSize ja ready toimivat määrityksen mukaisesti. Käytä jonotusstrategiaa antaaksesi tuottajalle paluukanavan:

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

Fetchin kanssa

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 välimuistia varten

Yleinen toimintamalli: lue runko kahdesti.

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

Response.clone() tekee tämän puolestasi ja palauttaa uuden Response-olion.

Huomioitavaa

  • Vain synkroniset tuottajat. pull(), joka tekee estävää PHP-I/O:ta, estää skriptin suorittamisen. Käytä fetch-siirtokoukkua HTTP varten; tiedostojen kohdalla lue ne mieluiten ensin muistiin. setTimeout / setInterval ovat käytettävissä (katso Tapahtumasilmukka), mutta moottori käyttää oletuksena virtuaaliaikaa — pull()-olion sisällä oleva setTimeout(cb, 1000) ei nukuta PHP, vaan siirtää silmukan kelloa eteenpäin. Reaaliaikainen tahdistus edellyttää, että isäntä ajaa tickEventLoop()-toimintoa seinäkellon mukaan.

Katso myös