Pereiti prie turinio

Srautai

Peržiūrėti kaip Markdown Redaguoti per GitHub Dokumentacija: 5b53f69

Phasis įgyvendina visą WHATWG Streams Standard. Kiekvieną klasę, kiekvieną skaitytuvo režimą ir kiekvieną pagalbinę funkciją. Tas pats kodas, kuris veikia Node ir šiuolaikinėse naršyklėse, veikia ir Phasis.

Tai įgyvendinimas, kurį fetch naudoja viduje — Response.body yra tikras ReadableStream, Request priima srautinius turinius, o tee(), reikalingas Response klonui sukurti, veikia.

Kas įtraukta

Klasė Būsena
ReadableStream Pilna
ReadableStreamDefaultController Pilna
ReadableByteStreamController Pilna
ReadableStreamDefaultReader Pilna
ReadableStreamBYOBReader Pilna
ReadableStreamBYOBRequest Pilna
WritableStream Pilna
WritableStreamDefaultController Pilna
WritableStreamDefaultWriter Pilna
TransformStream Pilna
TransformStreamDefaultController Pilna
ByteLengthQueuingStrategy Pilna
CountQueuingStrategy Pilna

Taip pat veikia dažniausiai naudojami metodai: pipeTo, pipeThrough, tee, getReader, asinchroninis iteravimas, Symbol.asyncIterator ir BYOB skaitymas naudojant { min }.

Duomenų pateikimas

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

Duomenų skaitymas

Trys skaitymo būdai:

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

Transformavimas

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

Baitų srautai (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 } palaikomas — read(view, { min: 2 }) išsprendžiamas tik tada, kai yra bent 2 elementai.

Atgalinis spaudimas

desiredSize ir ready veikia taip, kaip aprašyta specifikacijoje. Naudokite eilių strategiją, kad pateiktumėte duomenų gamintojui grįžtamąjį kanalą:

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

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

Dvigubas srautas spartinimui

Dažnas šablonas: perskaityti turinį du kartus.

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

Response.clone() tai atlieka už jus ir grąžina naują Response.

Apribojimai

  • Tik sinchroniniai duomenų gamintojai. pull(), atliekantis blokuojančias PHP įvesties ir išvesties operacijas, sustabdys scenarijų. HTTP naudokite Fetch transporto kablį; dirbant su failais, pirmiausia geriau įkelti juos į atmintį. Prieinami setTimeout / setInterval (žr. Įvykių ciklas), tačiau pagal numatytąsias nuostatas variklis naudoja virtualųjį laiką — pull() viduje esantis setTimeout(cb, 1000) neiš tiesų PHP, o pasuka ciklo laikrodį pirmyn. Kad veiktų realaus laiko tempas, pagrindinė aplinka turi valdyti tickEventLoop() pagal realų laikrodį.

Taip pat žr.