Перейти к содержимому

Потоки

Просмотреть как 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 });

Tee для кэширования

Распространённый сценарий: прочитать тело дважды.

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() по настенным часам.

См. также