Pāriet uz saturu

Plūsmas

Skatīt kā Markdown Rediģēt vietnē GitHub Dokumentācija vietnē 5b53f69

Phasis ievieš pilnu WHATWG Streams standartu. Katru klasi, katru lasītāja režīmu un katru palīgfunkciju. Tas pats kods, kas darbojas Node un mūsdienīgās pārlūkprogrammās, darbojas arī Phasis.

Šo realizāciju iekšēji izmanto fetch — Response.body ir īsts ReadableStream, Request pieņem straumējošas pamatdaļas, un darbojas tee(), kas nepieciešams Response klonēšanai.

Kas ir ieviests

Klase Statuss
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

Un arī bieži izmantotās metodes: pipeTo, pipeThrough, tee, getReader, asinhronā iterācija, Symbol.asyncIterator, BYOB lasīšana ar { min }.

Datu nodošana

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

Datu nolasīšana

Trīs nolasīšanas veidi:

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

Pārveidošana

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

Baitu plūsmas (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 } tiek atbalstīts — read(view, { min: 2 }) tiek atrisināts tikai tad, ja pieejami vismaz 2 elementi.

Atpakaļspiediens

desiredSize un ready darbojas, kā aprakstīts specifikācijā. Izmantojiet rindas stratēģiju, lai nodrošinātu producentam atgriezenisko kanālu:

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

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

Dublēšana kešatmiņai

Bieži izmantots paņēmiens: nolasīt pamattekstu divreiz.

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

Response.clone() to paveic jūsu vietā un atgriež jaunu Response.

Ierobežojumi

  • Tikai sinhroni producenti. pull(), kas veic bloķējošas PHP I/O darbības, bloķēs skriptu. Izmantojiet fetch transporta āķi HTTP; failiem ieteicams vispirms nolasīt saturu atmiņā. setTimeout / setInterval ir pieejami (skatiet Notikumu cikls), taču dzinis pēc noklusējuma izmanto virtuālo laiku — setTimeout(cb, 1000) pull() iekšienē neguļ PHP, bet gan pavirza cikla pulksteni uz priekšu. Reāllaika ritmu jānodrošina resursdatoram, vadot tickEventLoop() atbilstoši reālajam laikam.

Skatiet arī