Przejdź do treści

Strumienie

Wyświetl jako Markdown Edytuj w GitHub Dokumentacja: 5b53f69

Phasis implementuje pełny standard WHATWG Streams Standard. Każda klasa, każdy tryb czytnika, każdy pomocnik. Ten sam kod, który działa w Node i nowoczesnych przeglądarkach, działa także w Phasis.

Implementacja jest używana wewnętrznie przez fetch — Response.body jest prawdziwym ReadableStream, Request obsługuje strumieniowe treści, a tee() potrzebne do klonowania Response działa.

Co jest dostępne

Klasa Stan
ReadableStream Pełna
ReadableStreamDefaultController Pełna
ReadableByteStreamController Pełna
ReadableStreamDefaultReader Pełna
ReadableStreamBYOBReader Pełna
ReadableStreamBYOBRequest Pełna
WritableStream Pełna
WritableStreamDefaultController Pełna
WritableStreamDefaultWriter Pełna
TransformStream Pełna
TransformStreamDefaultController Pełna
ByteLengthQueuingStrategy Pełna
CountQueuingStrategy Pełna

Ponadto dostępne są często używane metody: pipeTo, pipeThrough, tee, getReader, iteracja asynchroniczna, Symbol.asyncIterator oraz odczyt BYOB z { min }.

Wytwarzanie

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

Odczyt

Trzy sposoby odczytu:

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

Przekształcanie

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

Strumienie bajtów (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 } jest obsługiwane — read(view, { min: 2 }) rozwiązuje się dopiero, gdy dostępne są co najmniej 2 elementy.

Kontrola przepływu wstecznego

desiredSize i ready działają zgodnie z opisem w specyfikacji. Użyj strategii kolejkowania, aby zapewnić producentowi kanał zwrotny:

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

Z 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 do buforowania

Typowy wzorzec: odczytaj treść dwukrotnie.

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

Response.clone() robi to za Ciebie i zwraca nowy Response.

Ograniczenia

  • Producenci wyłącznie synchroniczni. pull() wykonujący blokujące operacje wejścia/wyjścia PHP zablokuje skrypt. W przypadku HTTP użyj zaczepu transportu fetch; w przypadku plików lepiej wcześniej wczytać dane do pamięci. Dostępne są setTimeout / setInterval (zobacz Pętla zdarzeń), ale silnik domyślnie używa czasu wirtualnego — setTimeout(cb, 1000) wewnątrz pull() nie usypia PHP, lecz przesuwa zegar pętli. Aby uzyskać rzeczywiste tempo, host musi sterować tickEventLoop() względem zegara czasu rzeczywistego.

Zobacz także