Treci la conținut

Fluxuri

Vizualizați ca Markdown Editați pe GitHub Documentație la 5b53f69

Phasis implementează complet standardul WHATWG Streams Standard. Fiecare clasă, fiecare mod de citire, fiecare ajutor. Același cod care funcționează în Node și în browsere moderne funcționează și în Phasis.

Implementarea este cea folosită intern de fetch — Response.body este un ReadableStream real, Request acceptă corpuri transmise în flux, iar tee() necesar pentru a clona un Response funcționează.

Ce este inclus

Clasă Stare
ReadableStream Completă
ReadableStreamDefaultController Completă
ReadableByteStreamController Completă
ReadableStreamDefaultReader Completă
ReadableStreamBYOBReader Completă
ReadableStreamBYOBRequest Completă
WritableStream Completă
WritableStreamDefaultController Completă
WritableStreamDefaultWriter Completă
TransformStream Completă
TransformStreamDefaultController Completă
ByteLengthQueuingStrategy Completă
CountQueuingStrategy Completă

Plus metodele pe care le folosiți: pipeTo, pipeThrough, tee, getReader, iterație asincronă, Symbol.asyncIterator, citire BYOB cu { min }.

Producere

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

Consumare

Trei moduri de citire:

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

Transformare

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

Fluxuri de octeți (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 } este acceptat — read(view, { min: 2 }) se rezolvă numai când sunt disponibile cel puțin 2 elemente.

Contrapresiune

desiredSize și ready funcționează conform specificației. Folosiți o strategie de punere în coadă pentru a oferi producătorului un canal de răspuns:

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

Cu 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 pentru memorare în cache

Model uzual: citiți corpul de două ori.

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

Response.clone() face acest lucru în locul dvs. și returnează un Response nou.

Observații

  • Producători exclusiv sincroni. Un pull() care efectuează operații de I/O PHP cu blocare va bloca scriptul. Folosiți hook-ul de transport fetch pentru HTTP; pentru fișiere, citiți-le în memorie din timp. setTimeout / setInterval sunt disponibile (consultați Bucla de evenimente), dar motorul folosește implicit timp virtual — un setTimeout(cb, 1000) dintr-un pull() nu așteaptă PHP, ci avansează ceasul buclei. Pentru ritm în timp real, gazda trebuie să ruleze tickEventLoop() după ceasul de timp real.

Consultați și