Перейти до вмісту

Потоки

Переглянути як Markdown Редагувати на GitHub Документація для 5b53f69

Phasis реалізує повний стандарт потоків WHATWG Streams Standard. Кожен клас, кожен режим читання, кожен допоміжний засіб. Той самий код працює в Node, сучасних браузерах і Phasis.

Саме цю реалізацію fetch використовує всередині: Response.body — справжній ReadableStream, Request приймає потокові тіла, а tee(), потрібний для клонування Response, працює.

Що постачається

Клас Статус
ReadableStream Повний
ReadableStreamDefaultController Повний
ReadableByteStreamController Повний
ReadableStreamDefaultReader Повний
ReadableStreamBYOBReader Повний
ReadableStreamBYOBRequest Повний
WritableStream Повний
WritableStreamDefaultController Повний
WritableStreamDefaultWriter Повний
TransformStream Повний
TransformStreamDefaultController Повний
ByteLengthQueuingStrategy Повний
CountQueuingStrategy Повний

А також потрібні вам методи: pipeTo, pipeThrough, tee, getReader, асинхронна ітерація, Symbol.asyncIterator, читання BYOB за допомогою { min }.

Створення потоків

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

Читання потоків

Три способи читання:

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

Перетворення потоків

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

Байтові потоки (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 } підтримується — read(view, { min: 2 }) завершується, лише коли доступні щонайменше 2 елементи.

Зворотний тиск

desiredSize і ready працюють так, як описано у специфікації. Щоб передати виробнику сигнал зворотного зв’язку, скористайтеся стратегією черги:

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

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

Розгалуження для кешування

Поширений шаблон: прочитати тіло двічі.

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

Response.clone() зробить це за вас і поверне новий Response.

Застереження

  • Лише синхронні виробники. pull(), який виконує блокувальний PHP ввід-вивід, блокуватиме скрипт. Для HTTP використовуйте перехоплювач транспорту fetch; для файлів краще заздалегідь зчитати їх у пам’ять. Доступні setTimeout / setInterval (див. розділ Цикл подій), але за замовчуванням рушій використовує віртуальний час — setTimeout(cb, 1000) усередині pull() не присипляє PHP, а просуває годинник циклу подій. Для регулювання темпу реального часу хост має виконувати tickEventLoop() відповідно до реального годинника.

Див. також